package com.cdream.basic.mq.kafka;

import cn.hutool.core.util.ByteUtil;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import static com.cdream.basic.mq.kafka.KafkaConst.BROKER_LIST;
import static com.cdream.basic.mq.kafka.KafkaConst.TOPIC;

public class KafkaTtlProducer {
    public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        props.put("bootstrap.servers", BROKER_LIST);
        KafkaProducer<String,String> producer = new KafkaProducer<>(props);
        // Header中加入超时时间
        ProducerRecord<String, String> record1= new ProducerRecord<>(TOPIC,0,System.currentTimeMillis(),null,"msg_ttl1",new RecordHeaders().add(new RecordHeader("ttl",ByteUtil.longToBytes(20))));
        ProducerRecord<String, String> record2= new ProducerRecord<>(TOPIC,0,System.currentTimeMillis()-5000,null,"msg_ttl2",new RecordHeaders().add(new RecordHeader("ttl",ByteUtil.longToBytes(5))));
        ProducerRecord<String, String> record3= new ProducerRecord<>(TOPIC,0,System.currentTimeMillis(),null,"msg_ttl3",new RecordHeaders().add(new RecordHeader("ttl",ByteUtil.longToBytes(30))));

        producer.send(record1).get();
        producer.send(record2).get();
        producer.send(record3).get();
        producer.close();
    }
}
